package com.czk.springboot.reactive.reactivestream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * 底层：基于数据缓冲队列+消息驱动模型+异步回调机制
 * 编码：流式编程+链式调用+声明时API
 * 效果：优雅权益部+消息实时处理+高吞吐量+占用少量资源
 */
public class FlowProcessorDemo {
    /**
     * 自定义Processor--既是发布者也是订阅者
     * 继承发布者
     */
    static class MyProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {
        private Flow.Subscription subscription;

        @Override
        public void subscribe(Flow.Subscriber<? super String> subscriber) {
            super.subscribe(subscriber);
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            System.out.println("MyProcessor绑定完成");
            subscription.request(1);
        }

        @Override
        public void onNext(String item) {
            System.out.println("MyProcessor拿到数据，对数据进行加工" + item);
            String newItem = item + "czk";
            // 加工后的数据发出去
            super.submit(newItem);
            // 继续接收数据
            this.subscription.request(1);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onComplete() {
            System.out.println("MyProcessor接收到发布者的完成信号了");
        }
    }


    public static void main(String[] args) {
        // 1 定义一个发布者
        // SubmissionPublisher发布的所有数据在buffer区
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        // 2 中间操作处理器
        MyProcessor myProcessor = new MyProcessor();
        // 3 定义一个订阅者，订阅者·订阅发布者的数据
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            /**
             * d订阅时 在xxx事件发生时，执行回调
             *
             * @param subscription a new subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                System.out.println(Thread.currentThread().getName() + " 订阅开始了：" + subscription);
                // 从上游请求数据---可以指定请求数量  背压模式，自己控制自己的索要的元素数量
                this.subscription.request(1);
            }

            /**
             * 下一个元素到达时，执行回调，接收新数据
             *
             * @param item the item
             */
            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + " 接收到数据：" + item);
                // 被压模式，处理完后，自己再索要一个数据
                this.subscription.request(1);
            }

            /**
             * 错误发生时执行回掉
             *
             * @param throwable the exception
             */
            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + " 接收到错误信号：" + throwable);

            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + " subscriber 接收到完成信号：");
            }
        };

        // 绑定订阅者和发布者---发布者记住了所有的订阅者，有数据后把数据推送个订阅者--观察者模式
        publisher.subscribe(myProcessor);
        myProcessor.subscribe(subscriber);
        // 发布者发布消息
        for (int i = 0; i < 10; i++) {
            publisher.submit(String.valueOf(i));
        }
        //publisher.close();

        while (true) {
        }
    }
}
